package cn.itcast.function;

import cn.itcast.bean.SectorBean;
import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Put;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by JiaYang Xu
 * Date 2021/7/6 16:33
 * Description
 */
public class SectorPutHbaseWindowFunction implements AllWindowFunction<SectorBean, List<Put>, TimeWindow> {
    @Override
    public void apply(TimeWindow window, Iterable<SectorBean> values, Collector<List<Put>> out) throws Exception {
        //封装list
        List<Put> list = new ArrayList<>();
        for (SectorBean value : values) {
            Put put = new Put((value.getSectorCode() + value.getTradeTime()).getBytes());
            put.addColumn("info".getBytes(),"data".getBytes(), JSON.toJSONString(value).getBytes());
            list.add(put);
        }
        out.collect(list);
    }
}
